package com.atguigu.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.OrderWide;
import com.atguigu.bean.PaymentInfo;
import com.atguigu.bean.PaymentWide;
import com.atguigu.common.DWDStaticConstants;
import com.atguigu.common.DWMStaticConstants;
import com.atguigu.utils.DateUtil;
import com.atguigu.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Author:GaoFei
 * @Description:
 * @Date:Created in 18:11
 * @Modified By:
 */
public class PayWideApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        /**
         * 生产环境下一定要写的，不过学习阶段还要开启hdfs比较麻烦先关闭
         *         System.setProperty("HADOOP_USER_NAME","atguigu");
         *         env.setStateBackend(new FsStateBackend(OdsStaticConstants.CHECK_POINT_URL));
         *         // 设置checkpoint时间
         *         env.enableCheckpointing(5000L);
         *         // 设置精准一次性
         *         env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
         *         // 设置check超时时间
         *         env.getCheckpointConfig().setCheckpointTimeout(10000L);
         *         // 设置最多两个checkpoint一块
         *         env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
         *         // 设置两个check间隔
         *         env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
         */
        DataStreamSource<String> dataStreamSource = env.addSource(KafkaUtil.getKafkaConsumer(DWDStaticConstants.DWD_PAY_INFO, DWDStaticConstants.DWD_PAY_INFO_GROUP));
        SingleOutputStreamOperator<JSONObject> payWideParse = dataStreamSource.map(JSONObject::parseObject);
        DataStreamSource<String> orderWideDs = env.addSource(KafkaUtil.getKafkaConsumer(DWMStaticConstants.DWM_TOPIC_ORDER_WIDE_SINK, DWMStaticConstants.DWM_ORDER_WIDE_GROUP_ID));
        SingleOutputStreamOperator<JSONObject> orderWideParse = orderWideDs.map(JSONObject::parseObject);

        SingleOutputStreamOperator<JSONObject> payAppWithEventTs = payWideParse
                .assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner((element, recordTimestamp) -> DateUtil.getTime(element.getString("create_time"))));
        SingleOutputStreamOperator<JSONObject> orderWideWithEventTs = orderWideParse
                .assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner((element, recordTimestamp) -> DateUtil.getTime(element.getString("create_time"))));
        // 双流join
        SingleOutputStreamOperator<PaymentWide> result = payAppWithEventTs.keyBy(f -> f.getString("order_id"))
                .intervalJoin(orderWideWithEventTs.keyBy(f -> f.getString("order_id")))
                .between(Time.minutes(-15), Time.seconds(5))
                .process(new ProcessJoinFunction<JSONObject, JSONObject, PaymentWide>() {
                    @Override
                    public void processElement(JSONObject j1, JSONObject j2, Context context, Collector<PaymentWide> collector) throws Exception {
                        PaymentInfo paymentInfo = JSON.parseObject(j1.toJSONString(), PaymentInfo.class);
                        OrderWide orderWide = JSON.parseObject(j2.toJSONString(), OrderWide.class);
                        PaymentWide paymentWide = new PaymentWide(paymentInfo, orderWide);
                        collector.collect(paymentWide);
                    }
                });
        result.map(JSONObject::toJSONString)
                .addSink(KafkaUtil.getKafkaProducter(DWMStaticConstants.DWM_PAY_WIDE));
        result.print("payMentInfo>>>>");
        env.execute("PayWideApp");
    }
}
